package com.duowan.realtime.thrift;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.duowan.realtime.uniq.algorithm.bloomfilter.BloomFilter;

public class PartitionedBloomFilter {
	
	private final static Logger LOG = LoggerFactory.getLogger("bflog");
	
	private Map<String, BloomFilter> partitions = new ConcurrentHashMap<String, BloomFilter>();
	private String baseDir;
	public PartitionedBloomFilter() {
	}
	
	public Map<String, BloomFilter> getPartitons(){
		return partitions;
	}
	
	public void setBaseDir(String baseDir){
		this.baseDir = baseDir;
	}
	
	public void setPartitions(String partition, BloomFilter bloomFilter){
		partitions.put(partition, bloomFilter);
	}
	
	public PartitionedBloomFilter(String baseDir) {
		super();
		this.baseDir = baseDir;
	}


	public BloomFilter getBloomFilter(String partition) {
		BloomFilter r = partitions.get(partition);
		if(r == null) {
			synchronized (this) {
				try {
					r = loadBloomFilter(partition);
				} catch (Exception e) {
					throw new RuntimeException("error on load bloomfilter:"+partition,e);
				}
				partitions.put(partition, r);
			}
		}
		return r;
	}

	private BloomFilter loadBloomFilter(String partition) throws FileNotFoundException, IOException, ClassNotFoundException {
		File file = partitionFile(partition);
		if(file.exists()) {
			ObjectInputStream in = null;
			try {
				LOG.info("recover PartitionedBloomFilter, file:"+file);
				in = new ObjectInputStream(new BufferedInputStream(new FileInputStream(file)));
				return (BloomFilter)in.readObject();
			}finally {
				IOUtils.closeQuietly(in);
			}
		}else {
			return new BloomFilter(Integer.MAX_VALUE, 5);
		}
	}
	
	public void dump() {
		LOG.info("start to dump PartitionedBloomFilter. baseDir= "+ baseDir + ",partitions=" + partitions);
		for(Entry<String, BloomFilter> entry : partitions.entrySet()) {
			BloomFilter bf = entry.getValue();
			if(bf!= null && bf.isChange()) {
				try {
					doDump(entry.getKey(),bf);
					
					bf.cleanChange();
				} catch (IOException e) {
					LOG.error("error on dump partition=" + entry.getKey() + ",BloomFilter = "+ bf,e);
				}
			}
		}
	}
	
	private void doDump(String partition,BloomFilter bf) throws IOException {
		File file = partitionFile(partition);
		LOG.info("doDump PartitionedBloomFilter, partition:"+ partition + ",BloomFilter=" +bf +  ",bf size:" + bf.getNumberOfAddedElements() + ",file:"+file);
		file.getParentFile().mkdirs();
		
		ObjectOutputStream oos = new ObjectOutputStream(new BufferedOutputStream(new FileOutputStream(file)));
		try {
			oos.writeObject(bf);
		}finally {
			IOUtils.closeQuietly(oos);
		}
	}

	private File partitionFile(String partition) {
		File file = new File(baseDir+"/"+partition,"bloomfilter.dump");
		return file;
	}

	public void cleanNoChangePartitionsFromMemory() {
		for(String partition : partitions.keySet()) {
			BloomFilter bloomFilter = partitions.get(partition);
			if(!bloomFilter.isChange()){
				LOG.info("remove partition:" + partition + ",now partition size :" + partitions.size());
				partitions.remove(partition);
				bloomFilter = null ;
			}
		} 
	}
	
}
